package sgt.project;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class MakeUpInitBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector=collector;
    }

    @Override
    public void execute(Tuple input) {
        String []a=input.getStringByField("value").split(",");
        String event_time="1";
        String event_type="cart";
        String product_id="1";
        if (a.length>=3){
            if (a[0]!=null){
                event_time=a[0];
            }
            if(a[1]!=null){
                event_type=a[1];
            }
            if (a[2]!=null){
                product_id=a[2];
            }
        }
        System.out.println(a);
        collector.emit(new Values(event_time,event_type,product_id));
        collector.ack(input);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("event_time","event_type","product_id"));
    }

}
